package com.hulk.rpc.client;

import com.google.common.base.Preconditions;
import com.hulk.rpc.common.EventLoopGroupHolder;
import com.hulk.rpc.config.client.AppConfig;
import com.hulk.rpc.config.common.HostPort;
import com.hulk.rpc.config.common.ParseUtils;
import com.hulk.rpc.filter.ClientFilter;
import com.hulk.rpc.invoke.FailoverInvokerFactory;
import com.hulk.rpc.remote.RemoteException;
import com.hulk.rpc.remote.RemoteInterface;
import com.hulk.rpc.remote.RemoteServiceFactory;
import com.hulk.rpc.transport.client.App;
import io.netty.channel.EventLoopGroup;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.io.Closeable;
import java.io.IOException;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * @author xuxukang
 * @date 2019-03-10 09:28
 */
@Data
@Slf4j
public class HulkClient implements Closeable {

  private final ConcurrentHashMap<App, Boolean> appMap = new ConcurrentHashMap<>();
  private final RemoteServiceFactory remoteServiceFactory = new RemoteServiceFactory(new FailoverInvokerFactory());
  private final EventLoopGroup eventLoopGroup;
  private final CopyOnWriteArrayList<ClientFilter> filters = new CopyOnWriteArrayList<>();

  public HulkClient(String resource) {
    List<AppConfig> appConfigs = ParseUtils.getInstance().parseAppConfigs(resource);
    this.addConnects(appConfigs);
    eventLoopGroup = EventLoopGroupHolder.get();
  }

  public void addConnects(List<AppConfig> appConfigs) {
    appConfigs.parallelStream().forEach(this::addConnect);
  }

  public void addConnect(AppConfig appConfig) {
    Preconditions.checkNotNull(appConfig);
    App app = new App(appConfig, eventLoopGroup, filters);
    appMap.put(app, Boolean.TRUE);

  }

  public void addConnect(String group, String app, HostPort... hostPort) {
    Preconditions.checkNotNull(group);
    Preconditions.checkNotNull(app);
    Preconditions.checkNotNull(hostPort);

    AppConfig appConfig = new AppConfig().setApp(app)
            .setGroup(group);
    App serverApp = new App(appConfig, eventLoopGroup, filters);
    try {
      serverApp.setConnect(hostPort);
      appMap.put(serverApp, Boolean.TRUE);
    } catch (Exception e) {
      throw new RemoteException(e);
    }
  }

  public <T> T getService(Class<T> clazz) {
    return remoteServiceFactory.getService(clazz);
  }

  public <T> void registerService(Class<T> clazz) {
    Preconditions.checkNotNull(clazz);
    T service = remoteServiceFactory.getService(clazz);

    if (service != null) {
      return;
    }

    Optional<App> appOptional = appMap//
            .keySet()//
            .stream()//
            .filter(v -> v.isSupport(clazz))//
            .findFirst();

    if (appOptional.isEmpty()) {
      throw new RemoteException("not support this service, " + clazz.getName());
    }

    App app = appOptional.get();

    try {
      remoteServiceFactory.register(app, clazz);
    } catch (Exception e) {
      throw new RemoteException(e);
    }
  }

  public <T> void setFailover(Class<T> clazz, Object failover) {
    Preconditions.checkNotNull(failover);
    T service = remoteServiceFactory.getService(clazz);

    if (service == null) {
      throw new RemoteException("not register this service, " + clazz.getName());
    }

    try {
      App app = ((RemoteInterface) service).getApp();
      remoteServiceFactory.setFailover(app, clazz, service, failover);
    } catch (Exception e) {
      throw new RemoteException(e);
    }
  }

  @Override
  public void close() throws IOException {
    appMap.forEachKey(4, app -> {
      try {
        app.close();
      } catch (IOException e) {
        if (log.isWarnEnabled()) {
          log.warn("client close error", e);
        }
      }
    });

    EventLoopGroupHolder.release(eventLoopGroup);

    appMap.clear();

    remoteServiceFactory.close();
  }

  /**
   * 添加过滤器到最前面，最前面的会被最先执行
   */
  public void addFirst(ClientFilter filter) {
    filters.add(0, filter);
  }

  /**
   * 添加过滤器到最后面，最后面的会被最后执行
   */
  public void addLast(ClientFilter filter) {
    filters.add(filter);
  }
}
